use std::collections::HashMap;
use std::time::Duration;
use bincode::{Decode, Encode};
use bytes::{BufMut, BytesMut};
use serde::{Deserialize, Serialize};
use tokio::io::AsyncReadExt;
use tokio::net::TcpStream;
use crate::{RpcResult};
use crate::error::{CustomerErrorKind, RpcError};

pub static LOCAL_SEQ : &'static str = "localSeq";
pub static AUTH_CODE : &'static str = "authCode";

#[derive(Debug, Clone)]
pub struct RpcHeader {
    pub path: String,
    pub version: i32,
    pub status: i32,
    pub ext_data: RpcExtData,
}

pub struct RpcData<T> {
    pub header: RpcHeader,
    pub data: T,
}

#[derive(Debug, Clone)]
pub struct RpcExtData {
    pub ext_data_map: HashMap<String, String>,
}

impl RpcExtData {
    pub fn new(ext_data_map: HashMap<String, String>) -> RpcExtData {
        RpcExtData {
            ext_data_map,
        }
    }

    pub fn local_seq(&self) -> Option<String> {
        let result = self.ext_data_map.get(LOCAL_SEQ);
        match result {
            None => {None}
            Some(result) => {Some(result.to_string())}
        }
    }

    pub fn auth_code(&self) -> Option<String> {
        let result = self.ext_data_map.get(AUTH_CODE);
        match result {
            None => {None}
            Some(result) => {Some(result.to_string())}
        }
    }
}


impl RpcHeader {
    pub fn new(path: &'static str, local_seq: String, status: i32) -> RpcHeader {
        let mut map = HashMap::new();
        map.insert(LOCAL_SEQ.to_string(), local_seq);
        RpcHeader {
            path: path.to_string(),
            version: 1,
            status,
            ext_data: RpcExtData::new(map),
        }
    }

    pub fn response_header(&self, new_status: i32)-> RpcHeader {
        let mut ext_data = HashMap::new();
        if let Some(local_seq) = self.local_seq() {
           ext_data.insert(LOCAL_SEQ.to_string(), local_seq);
        }
        if let Some(auth_code) = self.auth_code() {
            ext_data.insert(AUTH_CODE.to_string(), auth_code);
        }
        RpcHeader {
            path: self.path.clone(),
            version: self.version.clone(),
            status: new_status,
            ext_data: RpcExtData::new(ext_data),
        }
    }

    pub fn local_seq(&self) -> Option<String> {
        let ext_data = &self.ext_data;
        let result = ext_data.ext_data_map.get(LOCAL_SEQ);
        match result {
            None => {None}
            Some(result) => {Some(result.to_string())}
        }
    }

    pub fn auth_code(&self) -> Option<String> {
        let ext_data = &self.ext_data;
        let result = ext_data.ext_data_map.get(AUTH_CODE);
        match result {
            None => {None}
            Some(result) => {Some(result.to_string())}
        }
    }
}

impl <T> RpcData<T> {
    pub fn new(path: &'static str, data: T) -> RpcData<T> {
        let header = RpcHeader {
            path: path.to_string(),
            version: 1,
            status: 0,
            ext_data: RpcExtData::new(HashMap::new()),
        };
        RpcData { header, data }
    }

    pub fn build(header: RpcHeader, data: T) -> RpcData<T> {
        RpcData { header, data }
    }
}

#[derive(Debug, Clone, Deserialize, Serialize, Encode, Decode)]
pub struct RpcPing;

#[derive(Debug, Clone, Deserialize, Serialize, Encode, Decode)]
pub struct RpcPong;

pub async fn read_packet_for_timeout(stream: &mut TcpStream, timeout: usize) -> RpcResult<Vec<u8>> {
    let data = tokio::time::timeout(Duration::from_millis((timeout) as u64),
                                    read_packet(stream)).await;
    match data {
        Ok(data) => {
            match data {
                Ok(data) => { Ok(data) }
                Err(error) => { Err(error.into()) }
            }
        }
        Err(_error) => {
            Err(RpcError::CustomerError{
                error_kid: CustomerErrorKind::TimeoutError,
                message: "读取数据超时".to_string(),
            })
        }
    }
}

pub async fn read_packet(stream: &mut TcpStream)-> RpcResult<Vec<u8>> {
    let mut buffer = [0; 256];
    let mut total_size = 0;
    loop {
        let bytes_read = stream.read(&mut buffer).await?;
        if bytes_read == 0 {
            let error = RpcError::CustomerError {
                error_kid: CustomerErrorKind::StreamClose,
                message: "连接关闭".to_string()
            };
            return Err(error);
        }
        total_size += bytes_read;
        if total_size > 1 {
            let length = buffer[0] as usize;
            if length == total_size -1  {
                return Ok(buffer[1..total_size].to_vec());
            }
        }
    }
}

pub fn str_to_packet(data: &str) -> Vec<u8> {
    let data_bytes = data.as_bytes();
    let mut buffer = BytesMut::with_capacity(data_bytes.len() + 1);
    buffer.put_u8(data_bytes.len() as u8);
    buffer.put_slice(data_bytes);
    buffer.freeze().to_vec()
}
